package com.orion.controller;

import cn.hutool.json.JSONUtil;
import com.orion.entity.TestMsg;
import com.orion.entity.User;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.*;

import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * @author Administrator
 * @date 2021/12/1
 */
@RestController
@RequestMapping("/msg")
public class MessageController {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    /**
     * 同步发送
     *
     * @param topic
     * @param msg
     * @throws ExecutionException
     * @throws InterruptedException
     */
    @GetMapping("/sendSimpleSync")
    public void sendSimpleSync(String topic, String key, TestMsg msg) throws ExecutionException, InterruptedException, TimeoutException {
        String jsonUser = JSONUtil.toJsonStr(msg);
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, key, jsonUser);
        SendResult<String, Object> sendResult = future.get(10, TimeUnit.SECONDS);
        System.out.println(sendResult);
    }

    @GetMapping("/sendSimple")
    public void sendSimple(String topic, String key, TestMsg msg) {
        String jsonUser = JSONUtil.toJsonStr(msg);
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, key, jsonUser);
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable ex) {
                System.out.println("send msg fail:" + ex);
            }

            @Override
            public void onSuccess(SendResult<String, Object> result) {
                System.out.println("send msg success:" + result.toString());
            }
        });
    }

    @GetMapping("/sendMoreSimple")
    public void sendMoreSimple(String topic, Integer times, String key, String sign, TestMsg msg) {
        for (int i = 0; i < times; i++) {
            msg.setName(sign + " : " + i);
            String jsonUser = JSONUtil.toJsonStr(msg);
            ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, (key != null ? key + i : null), jsonUser);
            future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
                @Override
                public void onFailure(Throwable ex) {
                    System.out.println("send msg fail:" + ex);
                }

                @Override
                public void onSuccess(SendResult<String, Object> result) {
                    System.out.println("send msg success:" + result.toString());
                }
            });
        }
    }

    @GetMapping("/sendSimpleToPartition")
    public void sendMoreSimple(String topic, String key, Integer partition, TestMsg msg) {
        String jsonUser = JSONUtil.toJsonStr(msg);
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, partition, key, jsonUser);
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable ex) {
                System.out.println("send msg fail:" + ex);
            }

            @Override
            public void onSuccess(SendResult<String, Object> result) {
                System.out.println("send msg success:" + result.toString());
            }
        });
    }

    @PostMapping("/sendUsers")
    public void sendUsers(@RequestBody List<User> users) {
        System.out.println(users);
        String topic="db";
        String key="dbKey";
        users.forEach(item -> {
            ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, key, JSONUtil.toJsonStr(item));
            future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
                @Override
                public void onFailure(Throwable ex) {
                    System.out.println("send msg fail:" + ex);
                }

                @Override
                public void onSuccess(SendResult<String, Object> result) {
                    System.out.println("send msg success:" + result.toString());
                }
            });
        });
    }
}
